use futures_util::future::join_all;

use std::sync::Arc;
use crate::consumer::{ConsumeLogs, ConsumeMetrics, ConsumeTraces};
use pdata::internal::data::protogen::collector::trace::v1::ExportTraceServiceRequest;
use pdata::internal::data::protogen::collector::metrics::v1::ExportMetricsServiceRequest;
use pdata::internal::data::protogen::collector::logs::v1::ExportLogsServiceRequest;

/// Fan-out consumer that sends traces to multiple downstream consumers concurrently.
pub struct FanoutTraces {
    downstream: Vec<Arc<dyn ConsumeTraces + Send + Sync + 'static>>,
}

impl FanoutTraces {
    /// Create a new fanout traces consumer with the given downstream consumers.
    pub fn new(downstream: Vec<Arc<dyn ConsumeTraces + Send + Sync + 'static>>) -> Self {
        Self { downstream }
    }
}

#[tonic::async_trait]
impl ConsumeTraces for FanoutTraces {
    async fn consume(
        &self,
        data: ExportTraceServiceRequest,
    ) {
        let futures = self
            .downstream
            .iter()
            .map(|c| c.consume(data.clone()));

        let _ = join_all(futures).await;
    }
}

/// Fan-out consumer that sends metrics to multiple downstream consumers concurrently.
pub struct FanoutMetrics {
    downstream: Vec<Arc<dyn ConsumeMetrics + Send + Sync + 'static>>,
}

impl FanoutMetrics {
    /// Create a new fanout metrics consumer with the given downstream consumers.
    pub fn new(downstream: Vec<Arc<dyn ConsumeMetrics + Send + Sync + 'static>>) -> Self {
        Self { downstream }
    }
}

#[tonic::async_trait]
impl ConsumeMetrics for FanoutMetrics {
    async fn consume(
        &self,
        data: ExportMetricsServiceRequest,
    ) {
        let futures = self
            .downstream
            .iter()
            .map(|c| c.consume(data.clone()));

        let _ = join_all(futures).await;
    }
}

/// Fan-out consumer that sends logs to multiple downstream consumers concurrently.
pub struct FanoutLogs {
    downstream: Vec<Arc<dyn ConsumeLogs + Send + Sync + 'static>>,
}

impl FanoutLogs {
    /// Create a new fanout logs consumer with the given downstream consumers.
    pub fn new(downstream: Vec<Arc<dyn ConsumeLogs + Send + Sync + 'static>>) -> Self {
        Self { downstream }
    }
}

#[tonic::async_trait]
impl ConsumeLogs for FanoutLogs {
    async fn consume(
        &self,
        data: ExportLogsServiceRequest,
    ) {
        let futures = self
            .downstream
            .iter()
            .map(|c| c.consume(data.clone()));

        let _ = join_all(futures).await;
    }
}


